package spring.boot.zk.service.impl;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;
import org.apache.curator.retry.RetryForever;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import spring.boot.zk.config.ZookeeperConfig;
import spring.boot.zk.service.ZookeeperService;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

@Service
public class ZookeeperServiceImpl implements ZookeeperService {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Override
    public boolean isExistNode(String path) {
        CuratorFramework client = ZookeeperConfig.getClient();
        client.sync();
        try {
            Stat stat = client.checkExists().forPath(path);
            return client.checkExists().forPath(path) != null;
        } catch (Exception e) {
            logger.error("isExistNode error...", e);
        }
        return false;
    }

    @Override
    public void createNode(CreateMode mode, String path) {
        CuratorFramework client = ZookeeperConfig.getClient();
        try {
            if (client.checkExists().forPath(path) != null){
                // 递归创建所需父节点
                client.create().creatingParentsIfNeeded().withMode(mode).forPath(path);
            }
        } catch (Exception e) {
            logger.error("createNode error...", e);
        }
    }

    @Override
    public void setNodeData(String path, String nodeData) {
        logger.info("设置节点信息 path={} nodeData={}",path,nodeData);
        CuratorFramework client = ZookeeperConfig.getClient();
        try {
            // 设置节点数据
            client.setData().forPath(path, nodeData.getBytes("UTF-8"));
        } catch (Exception e) {
            logger.error("setNodeData error...", e);

        }
    }

    @Override
    public void createNodeAndData(Integer mode, String path, String nodeData) {
        logger.info("创建节点信息并赋值 mode={} path={} nodeData={}",mode,path,nodeData);
        CuratorFramework client = ZookeeperConfig.getClient();
        try {
            if (client.checkExists().forPath(path) != null){
                logger.info("创建节点信息并赋值 path={} 已经存在，开始直接赋值",path);
                // 直接赋值
                client.setData().forPath(path, nodeData.getBytes("UTF-8"));
                return;
            }
            // 创建节点，关联数据
            client.create().creatingParentsIfNeeded().withMode(CreateMode.fromFlag(mode))
                    .forPath(path, nodeData.getBytes("UTF-8"));
        } catch (Exception e) {
            logger.error("createNode error...", e);
        }
    }

    @Override
    public String getNodeData(String path) {
        CuratorFramework client = ZookeeperConfig.getClient();
        try {
            // 数据读取和转换
            byte[] dataByte = client.getData().forPath(path);
            String data = new String(dataByte, "UTF-8");
            if (StringUtils.isNotEmpty(data)) {
                return data;
            }
        } catch (Exception e) {
            logger.error("getNodeData error...", e);
        }
        return null;
    }

    @Override
    public List<String> getNodeChild(String path) {
        CuratorFramework client = ZookeeperConfig.getClient();
        List<String> nodeChildDataList = new ArrayList<>();
        try {
            // 节点下数据集
            nodeChildDataList = client.getChildren().forPath(path);
        } catch (Exception e) {
            logger.error("getNodeChild error...", e);

        }
        return nodeChildDataList;
    }

    @Override
    public void deleteNode(String path, Boolean recursive) {
        CuratorFramework client = ZookeeperConfig.getClient();
        try {
            if (recursive) {
                // 递归删除节点
                client.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
            } else {
                // 删除单个节点
                client.delete().guaranteed().forPath(path);
            }
        } catch (Exception e) {
            logger.error("deleteNode error...", e);
        }
    }

    @Override
    public InterProcessReadWriteLock getReadWriteLock(String path) {
        // 写锁互斥、读写互斥
        InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(ZookeeperConfig.getClient(), path);
        return readWriteLock;
    }

    @Override
    public String watchPath(String path) {
        logger.info("开始监听节点信息: path={}",path);
        CuratorFramework client = ZookeeperConfig.getClient();
        try {
            byte[] content = client.getData().usingWatcher(new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    logger.info("监听器watchedEvent：" + watchedEvent);
                }
            }).forPath(path);
            logger.info("监听节点内容：" + new String(content));
            // 第一次变更节点数据
            client.setData().forPath(path, "new content".getBytes());
            // 第二次变更节点数据
            client.setData().forPath(path, "second content".getBytes());
            //  Thread.sleep(Integer.MAX_VALUE);
        } catch (Exception e) {
          //  client.close();
            logger.error("watchPath...", e);
        } /*finally {
            client.close();
        }*/
        return "success";
    }

    @Override
    public String pathChildCachePath(String path) {
        CuratorFramework client = ZookeeperConfig.getClient();
        try {
            //创建PathChildrenCache  参数：true代表缓存数据到本地
            PathChildrenCache pathChildrenCache = new PathChildrenCache(client, path, true);
            PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) {
                    ChildData data = event.getData();
                    switch (event.getType()) {
                        case CHILD_ADDED:
                            logger.info("子节点增加, path={}, data={}", data.getPath(), data.getData());
                            break;
                        case CHILD_UPDATED:
                            logger.info("子节点更新, path={}, data={}", data.getPath(), data.getData());
                            break;
                        case CHILD_REMOVED:
                            logger.info("子节点删除, path={}, data={}", data.getPath(), data.getData());
                            break;
                        case CONNECTION_RECONNECTED:
                            try {
                                System.out.println("Connection reconnected");
                                pathChildrenCache.rebuild();
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                            break;
                        case CONNECTION_SUSPENDED:
                            System.out.println("Connection suspended");
                            break;
                        case CONNECTION_LOST:
                            System.out.println("Connection lost");
                            break;
                        default:
                            break;
                    }
                }
            };
        } catch (Exception e) {
            logger.error("cachePath error...", e);
        }
        return "success";
    }


    @Override
    public String nodeCachePath(String path) {
        CuratorFramework client = ZookeeperConfig.getClient();
        try {
            // 数据是否压缩
            NodeCache nodeCache = new NodeCache(client, path, false);
            NodeCacheListener nodeCacheListener = new NodeCacheListener() {
                @Override
                public void nodeChanged() throws Exception {
                    ChildData currentData = nodeCache.getCurrentData();
                    logger.info("ZNode节点状态改变, path={} data={}", currentData.getPath(),new String(currentData.getData(), StandardCharsets.UTF_8));
                }
            };
            nodeCache.getListenable().addListener(nodeCacheListener);

            nodeCache.start();
        } catch (Exception e) {
            logger.error("cachePath error...", e);
        }
        return "success";
    }

    @Override
    public String treeCachePath(String path) {
        CuratorFramework client = ZookeeperConfig.getClient();
        try {
            TreeCache treeCache = new TreeCache(client, path);
            TreeCacheListener treeCacheListener = new TreeCacheListener() {
                @Override
                public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                    ChildData data = event.getData();
                    if (data != null) {
                        switch (event.getType()) {
                            case NODE_ADDED:
                                logger.info("[TreeCache]节点增加, path={}, data={}", data.getPath(), data.getData());
                                break;
                            case NODE_UPDATED:
                                logger.info("[TreeCache]节点更新, path={}, data={}", data.getPath(), data.getData());
                                break;
                            case NODE_REMOVED:
                                logger.info("[TreeCache]节点删除, path={}, data={}", data.getPath(), data.getData());
                                break;
                            default:
                                break;
                        }
                    } else {
                        logger.info("[TreeCache]节点数据为空, path={}", data.getPath());
                    }
                }
            };
            treeCache.getListenable().addListener(treeCacheListener);
            treeCache.start();
        } catch (Exception e) {
            logger.error("cachePath error...", e);
        }
        return "success";
    }

    @Override
    public String curatorNodeCachePath(String path) {
        CuratorFramework client = ZookeeperConfig.getClient();
        CuratorCache curatorCache = CuratorCache.builder(client, path).build();
        CuratorCacheListener listener = CuratorCacheListener
                .builder()
                .forNodeCache(new NodeCacheListener() {
                    @Override
                    public void nodeChanged() throws Exception {
                        //TODO ...
                    }
                })
                .build();
        curatorCache.listenable().addListener(listener);
        curatorCache.start();
        return "success";
    }

    @Override
    public InterProcessSemaphoreMutex lock(String path) {
        CuratorFramework client = ZookeeperConfig.getClient();
        return new InterProcessSemaphoreMutex(client, path);
    }

    @Override
    public InterProcessMutex mutexLock(String path) {
        CuratorFramework client = ZookeeperConfig.getClient();
        return new InterProcessMutex(client, path);
    }
    @Override
    public InterProcessReadWriteLock readWriteLock(String path){
        CuratorFramework client = ZookeeperConfig.getClient();
        return new InterProcessReadWriteLock(client, path);
    }

    @Override
    public Long counter(String path) {
        //创建节点
        CuratorFramework client = ZookeeperConfig.getClient();
        DistributedAtomicLong counter = new DistributedAtomicLong(client, path,new RetryForever(100));
        try {
            AtomicValue<Long> value  = counter.increment();
           return value.postValue();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public Long getCounter(String path) {
        //创建节点
        CuratorFramework client = ZookeeperConfig.getClient();
        DistributedAtomicLong counter = new DistributedAtomicLong(client, path,new RetryForever(100));
        try {
            AtomicValue<Long> value  = counter.get();
            return value.postValue();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
    /*@Override
    public String treeCachePath(String path) {
        CuratorFramework client = ZookeeperConfig.getClient() ;
        CuratorCache curatorCache = CuratorCache.builder(client, path).build();
        TreeCacheListener listener = TreeCacheListener
                .builder()
                .forTreeCache(new TreeCacheListener(){
                    @Override
                    public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {

                    }
                })
                .build();
        curatorCache.listenable().addListener(listener);
        curatorCache.start();
        return  "success";
    }

    @Override
    public String pathChildrenCachePath(String path) {
        CuratorFramework client = ZookeeperConfig.getClient() ;
        CuratorCache curatorCache = CuratorCache.builder(client, path).build();
        PathChildrenCacheListener listener = PathChildrenCacheListener
                .builder()
                .forPathChildrenCache(new PathChildrenCacheListener(){
                    @Override
                    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {

                    }
                })
                .build();
        curatorCache.listenable().addListener(listener);
        curatorCache.start();
        return  "success";
    }*/


}